package kafka.client.consumer;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KafkaConsumer {
	private static final Logger logger = LoggerFactory
			.getLogger(KafkaConsumer.class);
	private ConsumerConnector consumer;
	private String TOPIC;

	private static class SingletonHolder {
		private static final KafkaConsumer INSTANCE = new KafkaConsumer();
	}

	private KafkaConsumer() {
		initConsumer();
		registerExitEvent();
	}

	public static final KafkaConsumer getInstance() {
		return SingletonHolder.INSTANCE;
	}

	private final Properties CONFIG_PROPERTIES = new Properties();

	/*
	 * 加载配置文件，初始化Consumer
	 */
	private  void initConsumer() {
		try {
			String confFileName = System.getProperty("consumer.conf", "classpath:kafka-consumer-conf.properties");
			if (confFileName.startsWith("classpath:")) {
				confFileName = StringUtils.substringAfter(confFileName, "classpath:");
				CONFIG_PROPERTIES.load(KafkaConsumer.class.getClassLoader().getResourceAsStream(confFileName));
			} else {
				CONFIG_PROPERTIES.load(new FileInputStream(confFileName));
			}
			ConsumerConfig config = new ConsumerConfig(CONFIG_PROPERTIES);
			consumer = Consumer.createJavaConsumerConnector(config);
			this.TOPIC = CONFIG_PROPERTIES.getProperty("topic");
		} catch (IOException e) {
			logger.error("errors occurs when init the IMCProducer: \n",e);
			System.exit(1);
		}
	}

	/*
	 * 添加回调方法，关闭consumer
	 */
	private void registerExitEvent() {
		Runtime.getRuntime().addShutdownHook(new Thread() {
			public void run() {
				if (consumer != null)
					consumer.shutdown();
			}
		});
	}

	public void start() {
		if (StringUtils.isBlank(TOPIC)) {
			logger.error("TOPIC is null.");
			System.exit(1);
		}
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(TOPIC, new Integer(1));

		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
		StringDecoder valueDecoder = new StringDecoder(
				new VerifiableProperties());

		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer
				.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
		KafkaStream<String, String> stream = consumerMap.get(TOPIC).get(0);
		ConsumerIterator<String, String> it = stream.iterator();
		while (it.hasNext())
			logger.info(it.next().message());
		// System.out.println(it.next().message());
	}

	public static void main(String[] args) {
		// if(args.length < 1 || StringUtils.isBlank(args[0])){
		// logger.error("TOPIC is null,  USAGE: kafka.client.consumer.KafkaConsumer topic");
		// return;
		// }
		KafkaConsumer consumer = KafkaConsumer.getInstance();
		consumer.start();
	}

}
